package com.rrd.dw.canal.sink;

import java.io.IOException;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import com.alibaba.fastjson.JSON;

public class KafkaSink extends AbstractSink {
private KafkaProducer<String, String> producer =null;
private ProducerRecord<String, String> record =null;
private String topicName ="cannlTest";
public KafkaSink(String topicName){
	this.topicName =topicName;
	init();
}
private void init(){
	
	Properties properties = new Properties();
	try {
		properties.load(KafkaSink.class.getClassLoader().getResourceAsStream("conf.properties"));
//		properties.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
		properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("client.id", "pigeonCannlID");
        this.producer =new KafkaProducer<String, String>(properties); 
	} catch (IOException e) {
		e.printStackTrace();
	}
}

	@Override
	public void write(String key,String val) {
		// TODO Auto-generated method stub
		System.out.println(val);
		record = new ProducerRecord<String, String>(topicName,  key, val) ;
		producer.send(record);
	}
	@Override
	public void write(String key,Map<String,String> map) {
		System.out.println("############1");
		record = new ProducerRecord<String,String>(topicName,  key, JSON.toJSONString(map)) ;
		System.out.println("############2");
		producer.send(record);
	}
	@Override
	public void write(String val) {
		// TODO Auto-generated method stub
		System.out.println(val);
		record = new ProducerRecord<String, String>(topicName,  val) ;
		producer.send(record);
	}
	public void closeProducer(){
		
		if(this.producer !=null ){
			this.producer.close();
		}
	}
	
	
	public static void main(String[] args) throws IOException, InterruptedException, ExecutionException {
	 
	}
}
